package com.gjy.pulsar.disruptor;

import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.SubscriptionType;

/**
 * @author gjy
 * @version 1.0
 * @since 2025-08-06 11:10:07
 */
public class PulsarConsumerSetup {

    public static Consumer<byte[]> initPulsarConsumer(String serviceUrl, String topic,
                                                      String subscription, PulsarToDisruptorBridge bridge) throws PulsarClientException {

        PulsarClient client = PulsarClient.builder()
                .serviceUrl(serviceUrl)
                .build();

        return client.newConsumer()
                .topic(topic)
                .subscriptionName(subscription)
                .subscriptionType(SubscriptionType.Shared) // 关键：共享订阅
                .receiverQueueSize(100) // 控制预取量
                .messageListener((consumer, msg) -> {
                    bridge.forwardToDisruptor(msg); // 消息转交Disruptor
                }).subscribe();
    }

}
